package com.wz.jiangsu.wangteng.kafka.consumer;

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;


import java.util.List;

@Component
public class KafkaConsumer {

    //监听消费
    //批量消费
    @KafkaListener( topics = {"test_zzx"}, groupId = "test_zzx_groupA")
    public String onNormalMessage(ConsumerRecord<String, Object> record) {

        String result ;
        result = "主题：" + record.topic() + " 分区：" + record.partition() + " 生产者发出来的信息：" + record.value();
        System.out.println(result);
        return result;
    }

//    @KafkaListener(id = "consumer2", topics = {"test_zzx_wazuh"}, groupId = "test_zzx_groupA")
//    public String onNormalMessage2(ConsumerRecord<String, Object> record) {
//
//        String result ;
//        result = "主题：" + record.topic() + " 分区：" + record.partition() + " 生产者发出来的信息：" + record.value();
//        System.out.println(result);
//        return result;
//    }

    //批量消费
//    @KafkaListener(id = "consumer3", topics = {"sb_topic"}, groupId = "sb_group")
//    public void onBatchMessage(List<ConsumerRecord<String, Object>> records) {
//        System.out.println(">>> 批量消费一次，recoreds.size()=" + records.size());
//        for (ConsumerRecord<String, Object> record : records) {
//            System.out.println(record.value());
//        }
//    }


    // 将这个异常处理器的BeanName放到@KafkaListener注解的errorHandler属性里面
//    @KafkaListener(topics = {"sb_topice_ex_1"},errorHandler = "consumerAwareErrorHandler")
//    public void onMessage4(ConsumerRecord<?, ?> record) throws Exception {
//        throw new Exception("简单消费-模拟异常");
//    }
//
//    // 批量消费也一样，异常处理器的message.getPayload()也可以拿到各条消息的信息
//    @KafkaListener(topics = "sb_topic_ex_2",errorHandler="consumerAwareErrorHandler")
//    public void onMessage5(List<ConsumerRecord<?, ?>> records) throws Exception {
//        System.out.println("批量消费一次...");
//        throw new Exception("批量消费-模拟异常");
//    }



}
